package com.atguigu.flink.chapter11;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/12 23:54
 */
public class Flink16_UDF {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(line -> {
                    String[] split = line.split(",");
                    return new WaterSensor(split[0],
                            Long.parseLong(split[1]),
                            Integer.parseInt(split[2]));
                });


        Table table = tableEnv.fromDataStream(sensorDS);

        // 不注册直接用
//        table.select(call(MyLenth.class,$("id"))).execute().print();

        // 先注册
        tableEnv.createTemporaryFunction("mylen", MyLenth.class);

        // TableAPI
//        table.select($("id"),call("mylen",$("id"))).execute().print();

        // SQL
        tableEnv.sqlQuery("select id,mylen(id) from "+table).execute().print();



    }

    public static class MyLenth extends ScalarFunction{
        public int eval(String value){
            return value.length();
        }
    }
}
